package com.ecommerce.data

import com.ecommerce.utils.ConfigLoader
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._
import java.io.File

/**
 * 电商数据清洗类
 * 处理步骤：
 * - 读取HDFS中的原始CSV数据
 * - 处理缺失值、异常值
 * - 标准化日期格式
 * - 新增衍生字段（如交易金额）
 * - 同时保存到HDFS和本地项目目录
 */
object DataCleaner {
  // 配置参数
  private val config = ConfigLoader.loadConfig
  private val rawDataPath = config.getString("hadoop.raw.data")
  private val cleanDataPathHDFS = config.getString("hadoop.clean.data")

  // 本地项目输出路径（项目根目录下的output/clean_data）
  private val localOutputPath = new File("output/clean_data").getAbsolutePath

  /**
   * 执行数据清洗主流程
   * @param spark SparkSession实例
   */
  def clean(spark: SparkSession): Unit = {
    // 1. 读取原始数据
    val rawDF = readRawData(spark)

    // 2. 执行清洗逻辑
    val cleanedDF = processData(rawDF)

    // 3. 保存清洗后的数据（同时保存到HDFS和本地）
    saveCleanDataToHDFS(cleanedDF)
//    saveCleanDataToLocal(cleanedDF)

    println(s"数据清洗完成！")
    println(s"HDFS路径: $cleanDataPathHDFS")
    println(s"本地项目路径: $localOutputPath")
  }

  /**
   * 读取HDFS中的原始CSV数据
   */
  private def readRawData(spark: SparkSession): DataFrame = {
    val schema = StructType(Seq(
      StructField("InvoiceNo", StringType, nullable = false),
      StructField("StockCode", StringType, nullable = false),
      StructField("Description", StringType, nullable = true),
      StructField("Quantity", IntegerType, nullable = false),
      StructField("InvoiceDate", StringType, nullable = false),
      StructField("UnitPrice", DoubleType, nullable = false),
      StructField("CustomerID", StringType, nullable = true),
      StructField("Country", StringType, nullable = true)
    ))

    spark.read
      .option("header", "true")
      .option("sep", ",")
      .option("quote", "\"")
      .schema(schema)
      .csv(rawDataPath)
  }

  /**
   * 核心清洗逻辑
   */
  // 在processData方法中添加退货标记
  private def processData(rawDF: DataFrame): DataFrame = {
    rawDF
      .dropDuplicates()
      .filter(
        (col("Quantity") > 0 || col("InvoiceNo").like("%C%")) &&  // 允许退货订单（Quantity可能为负）
          (col("UnitPrice") > 0)
      )
      .filter(
        col("InvoiceNo").isNotNull &&
          col("StockCode").isNotNull &&
          col("InvoiceDate").isNotNull &&
          col("CustomerID").isNotNull
      )
      .withColumn("InvoiceDate", to_timestamp(col("InvoiceDate"), "M/d/yyyy H:mm"))
      .withColumn("TotalAmount", round(col("Quantity") * col("UnitPrice"), 2).cast(DecimalType(10, 2)))
      .withColumn("is_return", col("InvoiceNo").like("%C%"))  // 新增退货标记
      .select(
        "InvoiceNo", "StockCode", "Description", "Quantity", "InvoiceDate",
        "UnitPrice", "TotalAmount", "CustomerID", "Country", "is_return"  // 包含退货标记
      )
  }

  /**
   * 保存清洗后的数据到HDFS
   */
  private def saveCleanDataToHDFS(cleanedDF: DataFrame): Unit = {
    cleanedDF.write
      .option("header", "true")
      .option("sep", ",")
      .mode("overwrite")
      .csv(cleanDataPathHDFS)
  }
  
  /**
   * 保存清洗后的数据到本地项目目录（output/clean_data）
   */
//  private def saveCleanDataToLocal(cleanedDF: DataFrame): Unit = {
//    // 确保本地目录存在
//    val dir = new File(localOutputPath)
//    if (!dir.exists()) {
//      dir.mkdirs()  // 递归创建目录
//    }
//
//    // 合并为单个文件输出（避免多分区文件）
//    cleanedDF
//      .coalesce(1)  // 合并为1个分区
//      .write
//      .option("header", "true")
//      .option("sep", ",")
//      .mode("overwrite")
//      .csv(localOutputPath)
//  }
}
